package com.hex.ds.hdtp.core.inf.metadata.service.impl;

import com.hex.ds.hdtp.core.inf.common.enums.FieldType;
import com.hex.ds.hdtp.core.inf.common.utils.JdbcUtil;
import com.hex.ds.hdtp.core.inf.metadata.converter.IHiveSqlResultMapper;
import com.hex.ds.hdtp.core.inf.metadata.entity.TableFieldsEntity;
import com.hex.ds.hdtp.core.inf.metadata.po.SourceConfInfoPo;
import com.hex.ds.hdtp.core.inf.metadata.repository.ISourceConfInfoRepository;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;

import java.sql.Connection;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.sql.Statement;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * @Package com.hex.ds.hdtp.core.inf.metadata.service.impl
 * @ClassName HiveMethodService
 * @Description Hive元数据管理服务类
 * @Author gj.xu
 * @Date 2024/3/4 18:19
 * @Version v2.0
 **/
@Slf4j
@Service("HIVE")
public class HiveMethodService extends AbstractHiveQLMethodService {

    private final IHiveSqlResultMapper hiveSqlResultMapper;

    public HiveMethodService(ISourceConfInfoRepository sourceConfInfoRepository, IHiveSqlResultMapper hiveSqlResultMapper) {
        super(sourceConfInfoRepository);
        this.hiveSqlResultMapper = hiveSqlResultMapper;
    }

    @Override
    protected TableFieldsEntity toTableFieldsBean(Map<String, String> tableFieldsMap) {
        return hiveSqlResultMapper.toTableFieldsBean(tableFieldsMap);
    }

    @Override
    public List<TableFieldsEntity> tableStructure(String sourceCode, String schemaName, String tableName) {
        List<SourceConfInfoPo> sourceConfInfoPoList = getSourceConfInfoRepository().queryListBySourceCode(sourceCode);
        List<TableFieldsEntity> resultList = new ArrayList<>();
        String querySql = queryTableStructureSql(schemaName, tableName);
        try (Connection jdbcConnection = JdbcUtil.getJdbcConnection(sourceConfInfoPoList);
             Statement jdbcStatement = JdbcUtil.createJdbcStatement(jdbcConnection);
             ResultSet resultSet = JdbcUtil.executeQuerySql(jdbcStatement, querySql)
        ) {
            List<String> partitionColumns = getPartitionColumns(schemaName, tableName, jdbcConnection);
            resultList = processResultSet(resultSet, resultList, partitionColumns);
        } catch (SQLException e) {
            log.error("获取表结构失败", e.getStackTrace());
        }
        return resultList;
    }

    public List<TableFieldsEntity> tableStructureByType(String sourceCode, String schemaName, String tableName, String fieldType) {
        List<SourceConfInfoPo> sourceConfInfoPoList = getSourceConfInfoRepository().queryListBySourceCode(sourceCode);
        List<TableFieldsEntity> resultList = new ArrayList<>();
        String querySql = queryTableStructureSql(schemaName, tableName);
        try (Connection jdbcConnection = JdbcUtil.getJdbcConnection(sourceConfInfoPoList);
             Statement jdbcStatement = JdbcUtil.createJdbcStatement(jdbcConnection);
             ResultSet resultSet = JdbcUtil.executeQuerySql(jdbcStatement, querySql)
        ) {
            List<String> partitionColumns = getPartitionColumns(schemaName, tableName, jdbcConnection);
            resultList = processResultSet(resultSet, resultList, partitionColumns);
        } catch (SQLException e) {
            log.error("根据类型获取表结构失败", e.getStackTrace());
        }
        List<String> fieldTypes = FieldType.fromFieldType(fieldType);
        List<TableFieldsEntity> afterDealFields = resultList.stream().filter(e -> fieldTypes.contains(e.getColumnType().toUpperCase())).collect(Collectors.toList());
        return afterDealFields;
    }

    /**
     * @Method processResultSet <br>
     * @Param resultSet SQL查询的ResultSet对象
     * @Param resultList 要用字段信息填充的列表
     * @Return List<TableFieldsEntity>  <br>
     * @Description 将tableFieldsMap放入resultList中，如果出现null字段则停止<br>
     * @Author Yin.Yang<br>
     * @Date 2023/9/6 16:46 <br>
     * @Version V1.0<br>
     */
    private List<TableFieldsEntity> processResultSet(ResultSet resultSet, List<TableFieldsEntity> resultList, List<String> partitionColumns) throws SQLException {
        int columnId = 1; // 初始化自增列ID
        boolean foundNullValue = false; // 初始化找到null值的标志
        while (resultSet.next() && !foundNullValue) {
            Map<String, String> tableFieldsMap = new HashMap<>();
            if (extractTableFields(resultSet, tableFieldsMap, partitionColumns, "data_type")) {
                tableFieldsMap.put("columnId", String.valueOf(columnId));
                resultList.add(toTableFieldsBean(tableFieldsMap));
                columnId++;
            } else {
                foundNullValue = true;
            }
        }
        return resultList;
    }

    /**
     * @param schemaName 数据源中的库名
     * @param tableName  数据源中的表名
     * @Method getPartitionColumns <br>
     * @Return 分区字段的Set集合<String> <br>
     * @Description 从HIVE表中获取分区字段<br>
     * @Author Yin.Yang<br>
     * @Date 2023/9/12 11:38 <br>
     * @Version V1.0<br>
     */
    public List<String> getPartitionColumns(String schemaName, String tableName, Connection jdbcConnection) throws SQLException {
        List<String> partitionColumns = new ArrayList<>();
        boolean isHavePartitionColumns = false;
        String describeSql = String.format("DESCRIBE EXTENDED %s.%s ", schemaName, tableName);
        try (Statement stmt = jdbcConnection.createStatement();
             ResultSet resultSet = stmt.executeQuery(describeSql)) {
            while (resultSet.next()) {
                String colName = resultSet.getString("col_name");
                if (colName != null && colName.equals("# Partition Information")) {
                    isHavePartitionColumns = true;
                    break;
                }
            }
        }
        if (isHavePartitionColumns) {
            String querySql = this.getTablePartitionFieldSql(schemaName, tableName);
            try (
                    Statement stmt = jdbcConnection.createStatement();
                    ResultSet resultSet = stmt.executeQuery(querySql)
            ) {
                while (resultSet.next()) {
                    String partition = resultSet.getString("partition");
                    if (partition != null && partition.contains("=")) {
                        partitionColumns.add(partition.split("=")[0]);
                    }
                }
            } catch (SQLException e) {
                log.error("从HIVE表中获取分区字段失败", e.getStackTrace());
                throw e;
            }
        }
        return partitionColumns;
    }

}
